package com.warm.kafkalearn.kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class ProducerFastStart {
    private static final String brokerList = "localhost:9092";
    private static final String topic = "testKafka";
    private static final String transactionId = "transactionId";
    public static void main(String[] args) {
        Properties properties = new Properties();
        //设置key序列化器
        //properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //设置重试次数
        properties.put(ProducerConfig.RETRIES_CONFIG,10);
        //设置值序列化器
        //properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        //设置集群地址
        //properties.put("bootstrap.servers",brokerList);
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,brokerList);
        //配置事务
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,transactionId);
        //配置幂等性
        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);
        KafkaProducer<String,String> kafkaProducer = new KafkaProducer<String, String>(properties);
        //初始化事务
        kafkaProducer.initTransactions();
        //开启事务
        kafkaProducer.beginTransaction();
        try {
            //处理业务逻辑
            ProducerRecord<String,String> producerRecord1 = new ProducerRecord<>(topic,"message-1!");
            kafkaProducer.send(producerRecord1);

            ProducerRecord<String,String> producerRecord2 = new ProducerRecord<>(topic,"message-2!");
            kafkaProducer.send(producerRecord2);

            ProducerRecord<String,String> producerRecord3 = new ProducerRecord<>(topic,"message-3!");
            kafkaProducer.send(producerRecord3);

            kafkaProducer.commitTransaction();
        }catch (Exception e){
            kafkaProducer.abortTransaction();
        }
        /*ProducerRecord<String,String> producerRecord = new ProducerRecord<>(topic,"kafka-demo","Hello World!");
        try {
           *//* Future<RecordMetadata> send = kafkaProducer.send(producerRecord);
            RecordMetadata recordMetadata = send.get();
            System.out.println(recordMetadata.topic());
            System.out.println(recordMetadata.partition());
            System.out.println(recordMetadata.offset());*//*
            kafkaProducer.send(producerRecord, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e == null){
                        System.out.println(recordMetadata.topic());
                        System.out.println(recordMetadata.partition());
                        System.out.println(recordMetadata.offset());
                    }
                }
            });
        }catch (Exception e){
            e.printStackTrace();
        }*/
        kafkaProducer.close();
    }
}
